package com.yangchao.flow.domain;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

public class MyProcessor  extends SubmissionPublisher<String> implements Flow.Processor<String,String> {

    private Flow.Subscription subscription; //保存绑定关系

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        System.out.println("processor订阅绑定完成");
        this.subscription = subscription;
        subscription.request(1); //找上游要一个数据
    }

    @Override //数据到达，触发这个回调
    public void onNext(String item) {
        System.out.println("processor拿到数据："+item);
        //再加工
        item += "：哈哈";
        submit(item);//把我加工后的数据发出去
        subscription.request(1); //再要新数据
    }

    @Override
    public void onError(Throwable throwable) {

    }

    @Override
    public void onComplete() {

    }
}